package com.iov.failurelog.rabbitmq;

import com.alibaba.fastjson.JSON;
import com.iov.failurelog.domain.Location;
import com.iov.failurelog.utils.PoolDemoThread;
import com.iov.failurelog.utils.RedisUtil;
import com.iov.failurelog.vo.VehicleMsgEntity;
import com.rabbitmq.client.Channel;
import org.apache.avro.data.Json;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import static cn.hutool.core.thread.GlobalThreadPool.submit;

/**
 * @author Administrator
 * @version 1.0
 * @description: TODO
 * @date 2023/11/28 10:39
 */

@ServerEndpoint("/trapeze/trapeze")
@Component
public class LngLatMq {

    @Resource
    private PoolDemoThread poolDemoThread;

    private Session session;
    private static final Set<LngLatMq> connections = new CopyOnWriteArraySet<>();

    /**
    * @description: //TODO
    *  经纬度消费
    * @author Administrator
    * @date: 2023/11/28 10:54
    */
    @RabbitListener(queues = {"queue3"})
    public void trapezeMq(Message message, Channel channel) throws IOException {
        poolDemoThread.thread();
        submit(() -> {

            this.session = session;
            byte[] body = message.getBody();
            String msg = new String(body);
            // 解析
            int i = msg.indexOf("{");
            String substring = msg.substring(i);
            VehicleMsgEntity vehicleMsgEntity = JSON.parseObject(substring, VehicleMsgEntity.class);

            // redis 保存经纬度
            RedisTemplate redisTemplate = RedisUtil.redis;
            GeoOperations opsForGeo = redisTemplate.opsForGeo();
            Double longitude = Double.valueOf(vehicleMsgEntity.getLongitude());
            Double latitude = Double.valueOf(vehicleMsgEntity.getLatitude());
            opsForGeo.add("VehicleLGLA", new Point(longitude, latitude),"vehicle");

            Location location = new Location();
            location.setLng(longitude);
            location.setLat(latitude);
            String s = JSON.toJSONString(location);
            broadcast(s);
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
    /**
     * 当有ws 客户端连接进来，调用的方法
     * 处理连接事件
     * @param session
     */
    @OnOpen
    public void start(Session session) {
        this.session = session;
        System.out.println("this:" + this);
        String msg = String.format(session.toString());
        System.out.println(msg);

        connections.add(this);
//        broadcast(msg);
    }

    /**
     * 接收到消息的监听方法
     * client write    ------>   server  read
     * server write    ------>   client  read
     */
    @OnMessage
    public void incoming(String message) {
        session = this.session;
        String msg = String.format(message,session.toString());
        System.out.println("incoming: "+msg);
    }
    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void end(Session session) {
        connections.remove(this);
        String msg = String.format(session.toString());
//        broadcast(msg);
    }
    /**
     * 出错了
     * @param t
     * @throws Throwable
     */
    @OnError
    public void onError(Throwable t) throws Throwable {
        System.out.println("onError");
        t.printStackTrace();
    }

    /**
     * 广播消息
     * @param msg
     */
    private void broadcast(String msg) {
        for (LngLatMq client : connections) {
            try {
                synchronized (client) {
                client.session.getBasicRemote().sendText(msg);
                }
            } catch (IOException e) {
                connections.remove(client);
                try {
                    client.session.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
                e.printStackTrace();
            }
        }
        System.err.println(msg);
    }
}
